package main

import (
	"context"
	"encoding/json"
	"fmt"
	"log"
	"os"
	"time"

	"github.com/louloulin/dataflare/pkg/workflow/connector"
)

func main() {
	fmt.Println("=== DataFlare MySQL Connector Verification ===")

	// Check if MySQL credentials are provided
	mysqlHost := os.Getenv("MYSQL_HOST")
	mysqlPort := os.Getenv("MYSQL_PORT")
	mysqlUser := os.Getenv("MYSQL_USER")
	mysqlPassword := os.Getenv("MYSQL_PASSWORD")
	mysqlDatabase := os.Getenv("MYSQL_DATABASE")

	if mysqlHost == "" {
		mysqlHost = "localhost"
	}

	if mysqlPort == "" {
		mysqlPort = "3306"
	}

	if mysqlUser == "" || mysqlPassword == "" || mysqlDatabase == "" {
		fmt.Println("MySQL credentials not provided. Using mock mode.")
		verifyMySQLWithMocks()
		return
	}

	// Verify MySQL connector with real connection
	verifyMySQLWithRealConnection(mysqlHost, mysqlPort, mysqlUser, mysqlPassword, mysqlDatabase)
}

// Verify MySQL connector with mocks
func verifyMySQLWithMocks() {
	fmt.Println("\n=== Verifying MySQL Connector with Mocks ===")

	// Get connector factory
	factory := connector.DefaultRegistry.GetFactory()

	// Test MySQL source connector
	fmt.Println("Testing MySQL source connector...")
	mysqlSource, err := factory.CreateSourceConnector("mysql")
	if err != nil {
		log.Fatalf("Failed to create MySQL source connector: %v", err)
	}
	fmt.Printf("MySQL source connector created: %T\n", mysqlSource)

	// Initialize with mock configuration
	mockData := []map[string]interface{}{
		{"id": 1, "name": "Alice", "email": "alice@example.com"},
		{"id": 2, "name": "Bob", "email": "bob@example.com"},
		{"id": 3, "name": "Charlie", "email": "charlie@example.com"},
	}

	mockDataBytes, err := json.Marshal(mockData)
	if err != nil {
		log.Fatalf("Failed to marshal mock data: %v", err)
	}

	err = mysqlSource.Initialize(map[string]interface{}{
		"host":      "localhost",
		"port":      3306,
		"user":      "mock_user",
		"password":  "mock_password",
		"database":  "mock_db",
		"table":     "mock_table",
		"mock":      true,
		"mock_data": string(mockDataBytes),
	})
	if err != nil {
		log.Fatalf("Failed to initialize MySQL source connector: %v", err)
	}

	// Test MySQL sink connector
	fmt.Println("Testing MySQL sink connector...")
	mysqlSink, err := factory.CreateSinkConnector("mysql")
	if err != nil {
		log.Fatalf("Failed to create MySQL sink connector: %v", err)
	}
	fmt.Printf("MySQL sink connector created: %T\n", mysqlSink)

	// Initialize with mock configuration
	err = mysqlSink.Initialize(map[string]interface{}{
		"host":     "localhost",
		"port":     3306,
		"user":     "mock_user",
		"password": "mock_password",
		"database": "mock_db",
		"table":    "mock_table",
		"mock":     true,
	})
	if err != nil {
		log.Fatalf("Failed to initialize MySQL sink connector: %v", err)
	}

	// Start connectors
	ctx := context.Background()
	fmt.Println("Starting connectors...")
	err = mysqlSource.Start(ctx)
	if err != nil {
		log.Fatalf("Failed to start MySQL source connector: %v", err)
	}
	defer mysqlSource.Stop(ctx)

	err = mysqlSink.Start(ctx)
	if err != nil {
		log.Fatalf("Failed to start MySQL sink connector: %v", err)
	}
	defer mysqlSink.Stop(ctx)

	// Read from source
	fmt.Println("Reading from MySQL source...")
	batch, err := mysqlSource.Read(ctx)
	if err != nil {
		log.Fatalf("Failed to read from MySQL source: %v", err)
	}

	if batch == nil || len(batch.Records) == 0 {
		log.Fatalf("No records returned from MySQL source")
	}

	fmt.Printf("Read %d records from MySQL source\n", len(batch.Records))

	// Process records
	for i, record := range batch.Records {
		var data map[string]interface{}
		err := json.Unmarshal(record.Value, &data)
		if err != nil {
			log.Fatalf("Failed to parse record: %v", err)
		}

		fmt.Printf("Record %d: %v\n", i+1, data)

		// Add processed flag
		data["processed"] = true
		data["timestamp"] = time.Now().Format(time.RFC3339)

		// Update record
		updatedValue, err := json.Marshal(data)
		if err != nil {
			log.Fatalf("Failed to marshal record: %v", err)
		}
		record.Value = updatedValue
	}

	// Write to sink
	fmt.Println("Writing to MySQL sink...")
	err = mysqlSink.Write(ctx, batch)
	if err != nil {
		log.Fatalf("Failed to write to MySQL sink: %v", err)
	}

	// Flush sink
	err = mysqlSink.Flush(ctx)
	if err != nil {
		log.Fatalf("Failed to flush MySQL sink: %v", err)
	}

	fmt.Println("MySQL connector verification with mocks completed successfully!")
}

// Verify MySQL connector with real connection
func verifyMySQLWithRealConnection(host, port, user, password, database string) {
	fmt.Println("\n=== Verifying MySQL Connector with Real Connection ===")

	// Get connector factory
	factory := connector.DefaultRegistry.GetFactory()

	// Create test table
	fmt.Println("Creating test table...")
	// Note: In a real implementation, you would use a database library to create the table

	// Test MySQL source connector
	fmt.Println("Testing MySQL source connector...")
	mysqlSource, err := factory.CreateSourceConnector("mysql")
	if err != nil {
		log.Fatalf("Failed to create MySQL source connector: %v", err)
	}
	fmt.Printf("MySQL source connector created: %T\n", mysqlSource)

	// Initialize with real configuration
	err = mysqlSource.Initialize(map[string]interface{}{
		"host":     host,
		"port":     port,
		"user":     user,
		"password": password,
		"database": database,
		"query":    "SELECT 1 as id, 'test' as name",
	})
	if err != nil {
		log.Fatalf("Failed to initialize MySQL source connector: %v", err)
	}

	// Test MySQL sink connector
	fmt.Println("Testing MySQL sink connector...")
	mysqlSink, err := factory.CreateSinkConnector("mysql")
	if err != nil {
		log.Fatalf("Failed to create MySQL sink connector: %v", err)
	}
	fmt.Printf("MySQL sink connector created: %T\n", mysqlSink)

	// Initialize with real configuration
	err = mysqlSink.Initialize(map[string]interface{}{
		"host":     host,
		"port":     port,
		"user":     user,
		"password": password,
		"database": database,
		"table":    "test_output",
	})
	if err != nil {
		log.Fatalf("Failed to initialize MySQL sink connector: %v", err)
	}

	// Start connectors
	ctx := context.Background()
	fmt.Println("Starting connectors...")
	err = mysqlSource.Start(ctx)
	if err != nil {
		log.Fatalf("Failed to start MySQL source connector: %v", err)
	}
	defer mysqlSource.Stop(ctx)

	err = mysqlSink.Start(ctx)
	if err != nil {
		log.Fatalf("Failed to start MySQL sink connector: %v", err)
	}
	defer mysqlSink.Stop(ctx)

	// Read from source
	fmt.Println("Reading from MySQL source...")
	batch, err := mysqlSource.Read(ctx)
	if err != nil {
		log.Fatalf("Failed to read from MySQL source: %v", err)
	}

	if batch == nil || len(batch.Records) == 0 {
		log.Fatalf("No records returned from MySQL source")
	}

	fmt.Printf("Read %d records from MySQL source\n", len(batch.Records))

	// Process records
	for i, record := range batch.Records {
		var data map[string]interface{}
		err := json.Unmarshal(record.Value, &data)
		if err != nil {
			log.Fatalf("Failed to parse record: %v", err)
		}

		fmt.Printf("Record %d: %v\n", i+1, data)

		// Add processed flag
		data["processed"] = true
		data["timestamp"] = time.Now().Format(time.RFC3339)

		// Update record
		updatedValue, err := json.Marshal(data)
		if err != nil {
			log.Fatalf("Failed to marshal record: %v", err)
		}
		record.Value = updatedValue
	}

	// Write to sink
	fmt.Println("Writing to MySQL sink...")
	err = mysqlSink.Write(ctx, batch)
	if err != nil {
		log.Fatalf("Failed to write to MySQL sink: %v", err)
	}

	// Flush sink
	err = mysqlSink.Flush(ctx)
	if err != nil {
		log.Fatalf("Failed to flush MySQL sink: %v", err)
	}

	fmt.Println("MySQL connector verification with real connection completed successfully!")
}
